import logging
import os
import shutil
import tempfile
import uuid
import re
from typing import Tuple, List, Text, Set, Union, Optional, Iterable

from rasa.nlu.training_data import loading
from rasa.utils.io import DEFAULT_ENCODING

logger = logging.getLogger(__name__)


def get_core_directory(paths: Optional[Union[Text, List[Text]]],) -> Text:
    """从路径列表中递归收集所有core训练文件
    Args:
        paths: 所有的训练文件或包含训练文件的文件夹的路径列表
    Returns:
         包含所有找到的Core训练文件的临时目录的路径
    """
    core_files, _ = get_core_nlu_files(paths)
    return _copy_files_to_new_dir(core_files)


def get_nlu_directory(paths: Optional[Union[Text, List[Text]]],) -> Text:
    """从路径列表中递归收集所有NLU训练文件

    Args:
        paths: 所有的训练文件或包含训练文件的文件夹的路径列表
    Returns:
        包含所有找到的NLU训练文件的临时目录的路径
    """
    _, nlu_files = get_core_nlu_files(paths)
    return _copy_files_to_new_dir(nlu_files)


def get_core_nlu_directories(
    paths: Optional[Union[Text, List[Text]]],
) -> Tuple[Text, Text]:
    """从路径列表中递归收集所有训练文件

    Args:
        paths: 训练文件或包含它们的文件夹的路径列表。
    Returns:
        包含core文件的目录路径和包含NLU训练文件的目录路径。
    """

    story_files, nlu_data_files = get_core_nlu_files(paths)

    story_directory = _copy_files_to_new_dir(story_files)
    nlu_directory = _copy_files_to_new_dir(nlu_data_files)

    return story_directory, nlu_directory


def get_core_nlu_files(
    paths: Optional[Union[Text, List[Text]]]
) -> Tuple[List[Text], List[Text]]:
    """从有关的路径列表中递归收集所有训练文件.
    Args:
        paths: 训练文件或包含它们的文件夹的路径列表
    Returns:
        story和NLU文件路径组成各自的元组，sorted()之后准换为列表
    """

    story_files = set()
    nlu_data_files = set()

    if paths is None:
        paths = []
    elif isinstance(paths, str):
        paths = [paths]

    for path in set(paths):
        if not path:
            continue

        # 判断是文件路径，还是文件夹路径
        if _is_valid_filetype(path):
            if is_nlu_file(path):
                nlu_data_files.add(os.path.abspath(path))
            elif is_story_file(path):
                story_files.add(os.path.abspath(path))
        else:
            # 从文件夹获取：nlu文件、core类型文件
            new_story_files, new_nlu_data_files = _find_core_nlu_files_in_directory(path)
            story_files.update(new_story_files)
            nlu_data_files.update(new_nlu_data_files)

    return sorted(story_files), sorted(nlu_data_files)


def _find_core_nlu_files_in_directory(directory: Text,) -> Tuple[Set[Text], Set[Text]]:
    """
    从文件夹中获得nlu和core的训练数据
    :param directory:   包含训练数据的文件夹路径
    :return:
        story_files:    所有的story训练数据文件路径
        nlu_data_files: 所有的nlu训练数据文件路径
    """

    story_files = set()
    nlu_data_files = set()

    for root, _, files in os.walk(directory, followlinks=True):
        # 这里对文件进行排序，以确保顺序一致，以获得可重复的训练结果
        for f in sorted(files):
            full_path = os.path.join(root, f)

            if not _is_valid_filetype(full_path):
                continue

            if is_nlu_file(full_path):
                nlu_data_files.add(full_path)
            elif is_story_file(full_path):
                story_files.add(full_path)

    return story_files, nlu_data_files


def _is_valid_filetype(path: Text) -> bool:
    """训练文件验证：
    必须以.json或.md结尾，并且是有效文件"""

    is_file = os.path.isfile(path)
    is_datafile = path.endswith(".json") or path.endswith(".md")

    return is_file and is_datafile


def is_nlu_file(file_path: Text) -> bool:
    """检查文件是否是与Rasa兼容的nlu文件

    Args:
        file_path: 需要检查的文件路径。
    Returns: 如果是nlu文件，则为True；否则为False。
    """
    return loading.guess_format(file_path) != loading.UNK


def is_story_file(file_path: Text) -> bool:
    """检查文件是否是Rasa故事文件
    Args:
        file_path: 需要检查的文件路径
    Returns: 如果是故事文件，则为True；否则为False
    """

    if not file_path.endswith(".md"):
        return False

    try:
        with open(file_path, encoding=DEFAULT_ENCODING, errors="surrogateescape") as lines:
            return any(_contains_story_pattern(line) for line in lines)
    except Exception as e:
        # catch-all because we might be loading files we are not expecting to load
        logger.error(
            f"Tried to check if '{file_path}' is a story file, but failed to "
            f"read it. If this file contains story data, you should "
            f"investigate this error, otherwise it is probably best to "
            f"move the file to a different location. "
            f"Error: {e}"
        )
        return False


def _contains_story_pattern(text: Text) -> bool:
    """检查文件是否是story格式"""

    story_pattern = r".*##.+"

    return re.match(story_pattern, text) is not None


def is_domain_file(file_path: Text) -> bool:
    """检查给定的文件路径是否是Rasa域（domain）文件

    Args:
        file_path: 应该检查的文件的路径
    Returns:
        如果是域文件则为True，否则为False。
    """

    file_name = os.path.basename(file_path)

    return file_name in ["domain.yml", "domain.yaml"]


def is_config_file(file_path: Text) -> bool:
    """检查给定的文件路径是否是Rasa配置（config）文件

       Args:
           file_path: 应检查文件的路径
       Returns:
           如果是Rasa配置文件，则为True；否则为False
       """

    file_name = os.path.basename(file_path)

    return file_name in ["config.yml", "config.yaml"]


def _copy_files_to_new_dir(files: Iterable[Text]) -> Text:
    directory = tempfile.mkdtemp()
    for f in files:
        # 确保文件不会互相覆盖，因此前缀
        unique_prefix = uuid.uuid4().hex
        unique_file_name = unique_prefix + "_" + os.path.basename(f)
        shutil.copy2(f, os.path.join(directory, unique_file_name))

    return directory
